package com.example.couponservice.service.Impl;

import cn.hutool.core.bean.BeanUtil;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;

import com.example.couponservice.domain.dto.CouponDto;
import com.example.couponservice.domain.entity.Coupon;
import com.example.couponservice.domain.vo.couponOrderVo;
import com.example.couponservice.mapper.CouponMapper;
import com.example.couponservice.service.ICouponService;
import com.example.couponservice.utils.CurrentHoler;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;


/**
 * 优惠券服务实现类
 * 处理优惠券的发放、获取、订单创建等核心业务逻辑
 * 实现了分布式锁、Redis流处理等高级功能
 *
 * @author czwa
 */
@Service
@Slf4j
@EnableAsync
public class CouponServiceImpl extends ServiceImpl<CouponMapper, Coupon> implements ICouponService, ApplicationContextAware {

    @Autowired
    private CouponMapper couponMapper;
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private RedissonClient redissonClient;

    private ApplicationContext applicationContext;
    private ICouponService proxy;

    /**
     * 优惠券获取的Lua脚本
     * 用于保证优惠券获取操作的原子性
     * 脚本位于classpath:gainCoupon.lua
     */
    private static final DefaultRedisScript<Long> GAIN_COUPON_SCRIPT;
    static {
        GAIN_COUPON_SCRIPT = new DefaultRedisScript<>();
        GAIN_COUPON_SCRIPT.setLocation(new ClassPathResource("gainCoupon.lua"));
        GAIN_COUPON_SCRIPT.setResultType(Long.class);
    }

    /**
     * 订单处理线程池
     * 使用单线程池处理优惠券订单，保证处理顺序
     */
    private static final ExecutorService EXECUTOR_SERVICE = Executors.newSingleThreadExecutor();
    private volatile boolean running = true;

    /**
     * 设置Spring应用上下文
     * 用于获取服务代理对象
     *
     * @param applicationContext Spring应用上下文
     */
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
    }

    /**
     * 处理优惠券订单
     * 使用分布式锁保证同一用户同一优惠券的操作唯一性
     *
     * @param couponOrderVo 优惠券订单视图对象
     */
    private void handleCouponOrder(couponOrderVo couponOrderVo) {
        ICouponService serviceProxy = getProxy();
        if (serviceProxy == null) {
            log.error("代理对象未初始化，无法处理优惠券订单");
            return;
        }
        // 生成分布式锁，锁的键包含用户ID和优惠券ID
        RLock lock = redissonClient.getLock("lock:gain:coupon:" + couponOrderVo.getUserId() + ":" + couponOrderVo.getCouponId());
        try {
            // 尝试获取锁，最多等待10秒，持有锁最长30秒
            if (lock.tryLock(10, 30, TimeUnit.SECONDS)) {
                serviceProxy.createCouponOrder(couponOrderVo);
            } else {
                log.error("获取锁失败");
            }
        } catch (InterruptedException e) {
            log.error("锁获取被中断", e);
            Thread.currentThread().interrupt();
        } finally {
            // 只有当前线程持有锁时才释放，避免误释放
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }

    /**
     * 优惠券订单处理线程
     * 监听Redis流中的订单消息并处理
     */
    private class couponOrderHandler implements Runnable {
        @Override
        public void run() {
            while (running) {
                try {
                    // 从Redis流中读取消息
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
                    );

                    if (list == null || list.isEmpty()) {
                        // 无消息时休眠，避免CPU占用过高
                        TimeUnit.MILLISECONDS.sleep(100);
                        continue;
                    }

                    // 处理读取到的订单消息
                    MapRecord<String, Object, Object> entries = list.get(0);
                    Map<Object, Object> values = entries.getValue();
                    log.info("Stream entry values: {}", values);
                    couponOrderVo couponOrderVo = BeanUtil.fillBeanWithMap(values, new couponOrderVo(), true);
                    handleCouponOrder(couponOrderVo);
                    // 确认消息已处理
                    stringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", entries.getId());
                } catch (InterruptedException e) {
                    log.error("订单处理线程被中断");
                    Thread.currentThread().interrupt();
                    break;
                } catch (Exception e) {
                    log.error("优惠券获取异常", e);
                    handlerPendingList();
                }
            }
        }

        /**
         * 处理待处理的消息列表
         * 当处理主流程出现异常时，尝试处理pending-list中的消息
         */
        private void handlerPendingList() {
            while (running) {
                try {
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1),
                            StreamOffset.create("stream.orders", ReadOffset.from("0"))
                    );
                    if (list == null || list.isEmpty()) {
                        // 没有待处理消息时退出
                        break;
                    }

                    MapRecord<String, Object, Object> entries = list.get(0);
                    Map<Object, Object> values = entries.getValue();
                    couponOrderVo couponOrderVo = BeanUtil.fillBeanWithMap(values, new couponOrderVo(), true);
                    handleCouponOrder(couponOrderVo);
                    stringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", entries.getId());
                } catch (Exception e) {
                    log.error("处理pending-list异常", e);
                    // 异常后延迟处理，避免快速失败循环
                    try {
                        TimeUnit.SECONDS.sleep(500);
                    } catch (InterruptedException ex) {
                        Thread.currentThread().interrupt();
                    }
                }
            }
        }
    }

    /**
     * 初始化方法
     * 在服务启动时执行，启动订单处理线程
     */
    @PostConstruct
    private void init() {
        EXECUTOR_SERVICE.submit(new couponOrderHandler());
    }

    /**
     * 获取服务代理对象
     * 使用双重检查锁定确保线程安全
     *
     * @return ICouponService代理对象
     */
    private ICouponService getProxy() {
        if (proxy == null) {
            synchronized (this) {
                if (proxy == null) {
                    proxy = applicationContext.getBean(ICouponService.class);
                }
            }
        }
        return proxy;
    }

    /**
     * 添加优惠券到用户账户
     * 使用Lua脚本保证原子性操作
     *
     * @param id 优惠券ID
     * @throws RuntimeException 当优惠券已被抢光或重复获取时抛出
     */
    @Override
    public void addCouponToUser(Integer id) {
        // 确保代理对象已设置
        if (proxy == null) {
            proxy = applicationContext.getBean(ICouponService.class);
        }

        // 执行Lua脚本，传递优惠券ID和当前用户ID
        long result = stringRedisTemplate.execute(
                GAIN_COUPON_SCRIPT,
                Collections.emptyList(),
                id.toString(),
                CurrentHoler.getCurrentId().toString()
        );

        // 根据Lua脚本返回结果处理异常
        if (result != 0) {
            throw new RuntimeException(result == 1 ? "优惠券被抢光了" : "不能重复获取");
        }
    }

    /**
     * 销毁方法
     * 在服务关闭时执行，停止订单处理线程
     */
    @PreDestroy
    public void destroy() {
        running = false; // 停止循环
        EXECUTOR_SERVICE.shutdown();
        try {
            if (!EXECUTOR_SERVICE.awaitTermination(10, TimeUnit.SECONDS)) {
                EXECUTOR_SERVICE.shutdownNow();
            }
        } catch (InterruptedException e) {
            EXECUTOR_SERVICE.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }

    /**
     * 创建优惠券订单
     * 包含事务管理，确保操作原子性
     *
     * @param couponOrderVo 优惠券订单视图对象
     * @throws RuntimeException 当库存不足时抛出
     */
    @Transactional(rollbackFor = Exception.class)
    @Override
    public void createCouponOrder(couponOrderVo couponOrderVo) {
        UpdateWrapper<Coupon> updateWrapper = new UpdateWrapper<>();
        updateWrapper.eq("id", couponOrderVo.getCouponId())
                .gt("stock", 0)
                .setSql("stock = stock - 1");
        boolean result = update(updateWrapper);
        if (!result) {
            throw new RuntimeException("获取优惠券失败");
        }
        // 添加优惠券到用户
        couponMapper.addCouponToUser(couponOrderVo);
    }

    /**
     * 获取所有可用优惠券
     *
     * @param now 当前时间
     * @param currentId 当前用户ID
     * @return 可用优惠券列表
     */
    @Override
    public List<CouponDto> getAllCoupon(LocalDateTime now, Integer currentId) {
        return couponMapper.getAllCoupon(now, currentId);
    }
}